package com.zf.fork_join;

import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.*;
import java.util.stream.LongStream;

/**
 * ForkJoinPool线程池样例：
 * ForkJoinPool线程池：
 * 线程池中的每一个工作线程{@link java.util.concurrent.ForkJoinWorkerThread}都维护有一个 队列，每个工作线程不断获取并执行各自队列中的任务。
 * 当工作线程自己的队列中没有任务时，会主动的从其他线程的队列中 “偷取” 任务执行，这个 过程叫做“工作窃取”。
 * “工作窃取”机制保证了线程可以不停的执行任务直到全部完成，减少了等待任务的时间，可以充分利用CPU的资源，整体提高多线程任务的执行效率。
 * <p>
 * <p>
 * 普通的线程池：
 * 只有一个等待队列，线程池中的线程在完成任务后，会进入空闲状态，被动等待任务的到来，没有ForkJoinPool线程池高效
 */
public class ForkJoinPoolTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 手动创建一个 ForkJoinPool线程池，默认池中最大的活跃线程数量是 “处理器核心数(如果有超线程则 *2)”
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        // 使用系统提供的静态 ForkJoinPool线程池，默认池中最大的活跃线程数量是 “处理器核心数(如果有超线程则 *2) -1”
        // ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();

        // 记录用到的线程
        Set<String> threadRecordSet = new LinkedHashSet<>();

        // 使用 invoke 方法将任务交给ForkJoinPool线程池执行，并等待结果返回
        // MySumTask sumTask = new MySumTask(1, 100, threadRecordSet);
        // 阻塞等待结果
        // Long result = forkJoinPool.invoke(sumTask);
        // System.out.println(result);
        // printUsedThread(threadRecordSet);

        System.out.println("====================");

        // 使用 submit 方法将任务提交给 ForkJoinPool线程池 执行
        threadRecordSet.clear();
        MySumTask sumTask2 = new MySumTask(1, 100, threadRecordSet);
        ForkJoinTask<Long> submit2 = forkJoinPool.submit(sumTask2);

        // 关闭线程池，将线程池的shutdown标识设置为 true, 不再接收新的任务，继续完成已提交的任务
        // forkJoinPool.shutdown()不会阻塞当前调用线程
        forkJoinPool.shutdown();

        System.out.println("当前活跃线程数：" + forkJoinPool.getActiveThreadCount());

        long sum = LongStream.range(1, 101).sum();
        System.out.println("正确结果：" + sum);

        // 获取 submit2 的结果，阻塞获取
        Long result2 = submit2.get();
        System.out.println(result2);

        System.out.println("窃取任务数：" + forkJoinPool.getStealCount());
        // 打印一下用到的线程
        printUsedThread(threadRecordSet);
    }

    /**
     * 定义 ForkJoinTask，必须是 ForkJoinTask 类的子类才能被 ForkJoinPool 执行：
     * 定义一个递归执行的 RecursiveTask 任务，计算开始到结束范围内的所有正数的和;
     * RecursiveTask 是 ForkJoinTask的子类。
     */
    public static class MySumTask extends RecursiveTask<Long> {

        /**
         * 记录有哪些线程执行了这个任务
         */
        public final Set<String> threadRecordSet;

        private final int start;
        private final int end;

        public MySumTask(int start, int end, Set<String> threadRecordSet) {
            this.start = start;
            this.end = end;
            this.threadRecordSet = threadRecordSet;
        }


        @Override
        protected Long compute() {
            // 存储是哪个线程在执行任务
            threadRecordSet.add(Thread.currentThread().toString());

            // 设置最细的颗粒度，当到达最新的颗粒度时，不再拆分，执行计算
            if (end - start < 10) {
                try {
                    // 模拟耗时
                    TimeUnit.MILLISECONDS.sleep(250);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                long res = 0;
                for (int i = start; i <= end; i++) {
                    res += i;
                }
                return res;
            }
            // 没有到达最小的颗粒度，则继续拆分
            int middle = (start + end) / 2;
            MySumTask leftTask = new MySumTask(start, middle, threadRecordSet);
            MySumTask rightTask = new MySumTask(middle + 1, end, threadRecordSet);
            // 执行fork()操作，将这两个任务交给“执行当前任务的ForkJoinPool线程池”中 异步执行，会调用 这两个任务的compute()方法，重复compute()中的操作
            leftTask.fork();
            rightTask.fork();
            // 等待这两个任务执行完毕并获取结果
            return leftTask.join() + rightTask.join();
        }
    }


    /**
     * 打印使用到的线程
     */
    public static void printUsedThread(Set<String> threadRecordSet) {
        System.out.println("使用到的线程数量：" + threadRecordSet.size());
        for (String ts : threadRecordSet) {
            System.out.println(ts);
        }
    }
}
